package com.tinem.platform.service.message.mqtt;

import cn.hutool.core.util.ReUtil;
import com.alibaba.fastjson.JSONObject;
import com.tinem.platform.module.pojo.co.RedisKeyEnum;
import com.tinem.platform.module.pojo.enums.message.MqttMessageType;
import com.tinem.platform.module.starter.EmqxClient;
import com.tinem.platform.module.starter.MqttMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author fengzhihao
 * @version v1
 * @program: platform
 * @className UserOnlineMqttMessageListener
 * @description TODO
 * @site
 * @company
 * @create 2022/1/19 15:02:14
 */
@Slf4j
@Component
public class UserOnlineMqttMessageListener implements MqttMessageListener {

    @Value("${spring.profiles.active}")
    protected String active;

    @Resource
    EmqxClient emqxClient;

    @Resource
    StringRedisTemplate stringRedisTemplate;

    @Override
    public String topicFilter() {
        return "$SYS/brokers/#";
    }

    @Override
    public void messageArrived(String clientId,String topic, MqttMessage message) {
        String body = new String(message.getPayload());
        if(ReUtil.contains("\\$SYS/brokers/[a-zA-Z@0-9\\.]+/clients/[a-zA-Z@0-9\\.\\_]+_[0-9]+/connected",topic)){
            String eventClientId = topic.split("/")[4];
            eventClientId = eventClientId.substring(0,eventClientId.lastIndexOf("_"));
            userOnline(clientId,eventClientId,body);
        }else if(ReUtil.contains("\\$SYS/brokers/[a-zA-Z@0-9\\.]+/clients/[a-zA-Z@0-9\\.\\_]+_[0-9]+/disconnected",topic)){
            String eventClientId = topic.split("/")[4];
            eventClientId = eventClientId.substring(0,eventClientId.lastIndexOf("_"));
            userOffline(clientId,eventClientId,body);
        }
    }
    public void userOnline(String clientId,String eventClientId,String message){
        JSONObject onLineMessage = JSONObject.parseObject(message);

        //防止事件重复处理。
        String eventId = RedisKeyEnum.mqtt_user_online_event.getKey(onLineMessage.getString("clientid"),onLineMessage.getString("connected_at"));
        Long eventNum = stringRedisTemplate.opsForValue().increment(eventId);
        stringRedisTemplate.expire(eventId,5, TimeUnit.MINUTES);
        //如果不是第一个处理此事件，则不处理
        if(eventNum != 1){
            return;
        }

        String userId = onLineMessage.getString("username");
        String key = RedisKeyEnum.mqtt_user_online.getKey(userId);
        Long sessinNum = stringRedisTemplate.opsForValue().increment(key);
        stringRedisTemplate.expire(key,5, TimeUnit.HOURS);
        log.debug("user online num:({}):({})",sessinNum,key);
        if(sessinNum == 1){
            //用户上线
            log.info("user({}) online",userId);
            emqxClient.sendMessageTopic(clientId, MqttMessageType.USER_ONLINE.getTopic(active,eventClientId),userId);
        }

    }

    public void userOffline(String clientId,String eventClientId,String message){
        JSONObject onLineMessage = JSONObject.parseObject(message);

        //防止事件重复处理。
        String eventId = RedisKeyEnum.mqtt_user_online_event.getKey(onLineMessage.getString("clientid"),onLineMessage.getString("disconnected_at"));
        Long eventNum = stringRedisTemplate.opsForValue().increment(eventId);
        stringRedisTemplate.expire(eventId,5, TimeUnit.MINUTES);
        //如果不是第一个处理此事件，则不处理
        if(eventNum != 1){
            return;
        }

        String userId = onLineMessage.getString("username");
        String key = RedisKeyEnum.mqtt_user_online.getKey(userId);
        Long sessinNum = stringRedisTemplate.opsForValue().decrement(key);
        log.debug("user offline num:({}):({})",sessinNum,key);
        if(sessinNum <= 0){
            //用户下线
            stringRedisTemplate.delete(key);
            log.info("user({}) offline",userId);
            emqxClient.sendMessageTopic(clientId, MqttMessageType.USER_OFFLINE.getTopic(active,eventClientId),userId);
        }
    }
}
